import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.simp.SimpMessagingTemplate;

import java.util.Arrays;
import java.util.Properties;

public class WsMessageThread implements Runnable {
    private MessageChannel messageChannel;
    private SimpMessagingTemplate wsTemplate;
    private KafkaConsumer<Object, Object> consumer;

    public WsMessageThread(MessageChannel messageChannel, MessageConverter messageConverter) {
        this.messageChannel = messageChannel;
        this.wsTemplate = new SimpMessagingTemplate(messageChannel);
        wsTemplate.setMessageConverter(messageConverter);
        String serverAddress = "localhost:9092";
        String topicName = "day-count-result";
        Properties defaultConsumerProps = new Properties();
        defaultConsumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, serverAddress);
        defaultConsumerProps.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 2000);
        defaultConsumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);
        defaultConsumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "day-count");
        defaultConsumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        defaultConsumerProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
        defaultConsumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        defaultConsumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
        consumer = new KafkaConsumer<>(defaultConsumerProps);
        consumer.subscribe(Arrays.asList(topicName));
    }

    @Override
    public void run() {
        while (true) {
            try {
                Thread.sleep(1000);
                ConsumerRecords<Object, Object> records = consumer.poll(1000);
                if (!records.isEmpty()){
                    for (ConsumerRecord<Object, Object> record : records) {
                        Greeting greeting = new Greeting();
                        greeting.setDate(record.key().toString());
                        greeting.setValue(record.value().toString());
                        greeting.setContent(record.key().toString().replaceAll("%%","  ")+"  "+record.value().toString());
                        wsTemplate.convertAndSend("/topic/greetings", greeting);
                        System.out.println(record.offset() + "  " + record.key().toString() + "  " + record.value());
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }

        }
    }
}
